"""
Case Type   : 发布订阅
Case Name   : 发布端使用gs_probackup远程备份
Create At   : 2022-11-07
Owner       : opentestcase026
Description :
    1.在两个集群创建表
    2.创建发布订阅
    3.发布端备节点远程全量备份主节点
    4.发布端插入数据
    5.订阅端查询数据是否更新
    6.修改发布端
    7.发布端增量备份
    8.使用全量备份恢复远程恢复发布端主节点
    9.发布端插入表数据
    10.订阅端查询数据是否更新
    11.使用增量备份恢复远程恢复发布端主节点
    12.发布端插入表数据
    13.订阅端查询数据是否更新
    14.发布端删除表数据
    15.订阅端查询表数据
    16.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.数据更新
    6.成功
    7.成功
    8.成功
    9.成功
    10.数据更新
    11.成功
    12.成功
    13.表1更新,表2不更新
    14.成功
    15.表1更新,表2不更新
    16.成功
History     :
    modified：2023-2-6 by opentestcase033:优化备份更改目录权限步骤，修改清理环境
    步骤中断言信息
"""

import os
import time
import unittest
from yat.test import macro
from yat.test import Node
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != Primary_SH.get_node_num(), '非1+2环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("-----------this is setup-----------")
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_pub_sta = Node(node='Standby1DbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_pub_sta = CommonSH('Standby1DbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.comsh_sta_pub = []
        self.nodelist = ['Standby1DbUser', 'Standby2DbUser']
        for i in range(2):
            self.comsh_sta_pub.append(CommonSH(self.nodelist[i]))
        self.tb_name1 = 't_pub_sub_0066_1'
        self.tb_name2 = 't_pub_sub_0066_2'
        self.subname = "s_pub_sub_0066"
        self.pubname = "p_pub_sub_0066"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.backup_path = os.path.join(self.parent_path_pub,
                                        'back_pub_sub_case066')
        self.remote_path = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'app/bin')
        self.remote_libpath = os.path.join(
            os.path.dirname(macro.DB_INSTANCE_PATH), 'app/lib')
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.enable_cbm_tracking = self.com_pub.show_param(
            "enable_cbm_tracking")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
            f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
            f'-W {self.pri_userdb_sub.db_password}'
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(result)
        cmd = f"cp " \
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')}" \
            f" {os.path.join(self.parent_path_sub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(result)
        self.hostname = self.pri_userdb_sub.sh('hostname').result().strip()
        text = "--step:预置条件,修改pg_hba expect:成功"
        self.log.info(text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload ', self.constant.GSGUC_SUCCESS_MSG,
            'enable_cbm_tracking=on')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        result = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        result = self.commsh_pub.restart_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in result or 'Degrade' in result
        self.assertTrue(flg, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host all  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)

    def test_pub_sub(self):
        text = "--step1:在两个集群创建表 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop table if exists {self.tb_name1};" \
            f"drop table if exists {self.tb_name2};" \
            f"create table {self.tb_name1}(id1 int primary key,id2 int,id3 " \
            f"int);" \
            f"create table {self.tb_name2}(id int primary key constraint " \
            f"id_nn not null,use_filename varchar2(20),filename " \
            f"varchar2(255),text varchar2(2000))" \
            f"partition by range(id)" \
            f"( " \
            f"      partition p1 values less than(30)," \
            f"      partition P2 values less than(60)," \
            f"      partition P3 values less than(90)," \
            f"      partition P4 values less than(MAXVALUE)" \
            f");"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.DROP_TABLE_SUCCESS),
                         2, '执行失败' + text)
        self.assertEqual(pub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         4, '执行失败' + text)
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertEqual(sub_result.count(self.constant.DROP_TABLE_SUCCESS),
                         2, '执行失败' + text)
        self.assertEqual(sub_result.count(self.constant.TABLE_CREATE_SUCCESS),
                         4, '执行失败' + text)

        text = "--step2:创建发布订阅 expect:成功--"
        self.log.info(text)
        sql_cmd = f"drop publication if exists {self.pubname};" \
            f"create publication {self.pubname} for table " \
            f"{self.tb_name1},{self.tb_name2} ;"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        self.assertIn(self.constant.drop_pub_succ_msg, pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_pub_succ_msg, pub_result,
                      '执行失败' + text)
        result = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        sql_cmd = f"drop subscription if exists {self.subname};" \
            f"create subscription {self.subname} connection " \
            f"'host={self.pri_userdb_pub.db_host} " \
            f"port={self.port} user={self.pri_userdb_pub.db_user} " \
            f"dbname={self.pri_userdb_pub.db_name} " \
            f"password={self.pri_userdb_pub.ssh_password}' publication " \
            f"{self.pubname};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn(self.constant.drop_sub_succ_msg, sub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.create_sub_succ_msg, sub_result, '执行失败'
                      + text)

        text = "--step3:发布端备节点远程全量备份主节点 expect:成功--"
        self.log.info(text)
        check_cmd = f'if [ -d {self.backup_path} ]; ' \
            f'then echo "exists"; else echo "not exists"; fi'
        self.log.info(check_cmd)
        result = self.pri_userdb_pub_sta.sh(check_cmd).result()
        self.log.info(result)
        if result == 'exists':
            sql_cmd = f"rm -rf {self.backup_path};"
            result = self.pri_userdb_pub_sta.sh(sql_cmd).result()
            self.log.info(result)
        sql_cmd = f"mkdir {self.backup_path};chmod 700 {self.backup_path}"
        self.log.info(sql_cmd)
        result = self.pri_userdb_pub_sta.sh(sql_cmd).result()
        self.log.info(result)
        result = self.commsh_pub_sta.exec_pro_backup_init(self.backup_path)
        self.log.info(result)
        self.assertTrue(result, '执行失败' + text)
        result = self.commsh_pub_sta.exec_pro_back_add(
            self.backup_path, 'instance2', 
            other_cmd=f"--remote-proto=ssh "
            f"--remote-host={self.pri_userdb_pub.db_host} --remote-port=22 "
            f"--remote-user={self.pri_userdb_pub.ssh_user} "
            f"--remote-path={self.remote_path} "
            f"--remote-libpath={self.remote_libpath}")
        self.log.info(result)
        self.assertTrue(result, '执行失败' + text)
        result = self.commsh_pub_sta.exec_pro_backup_backup(
            self.backup_path, 'instance2', 'full', self.pri_userdb_pub.db_name,
            other_cmd=f"-U {self.pri_userdb_pub.db_user} -W "
            f"{self.pri_userdb_pub.ssh_password} --remote-proto=ssh "
            f"--remote-host={self.pri_userdb_pub.db_host} "
            f"--remote-port=22 --remote-user={self.pri_userdb_pub.ssh_user} "
            f"--remote-path={self.remote_path} "
            f"--remote-libpath={self.remote_libpath} --backup-pg-replslot "
            f"-j 16")
        self.log.info(result)
        self.assertTrue(result, '执行失败' + text)

        text = "--step4:发布端插入数据 expect:成功--"
        self.log.info(text)
        pub_result = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name1} values(2,1,1);"
            f"insert into {self.tb_name2} values(1,'first','%一','');")
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         '执行失败' + text)

        text = "--step5:订阅端查询数据是否更新 expect:数据更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);" \
            f"select * from {self.tb_name1};select * from {self.tb_name2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('2 |   1 |   1', sub_result, '执行失败' + text)
        self.assertIn('1 | first        | %一', sub_result, '执行失败' + text)

        text = "--step6:修改发布端 expect:成功--"
        self.log.info(text)
        pub_result = self.commsh_pub.execut_db_sql(
            f"alter publication {self.pubname} drop table {self.tb_name2}; ")
        self.log.info(pub_result)
        self.assertIn(self.constant.alter_pub_succ_msg, pub_result, '执行失败'
                      + text)

        text = "--step7:发布端增量备份 expect:成功--"
        self.log.info(text)
        result = self.commsh_pub_sta.exec_pro_backup_backup(
            self.backup_path, 'instance2', 'PTRACK --stream',
            self.pri_userdb_pub.db_name, other_cmd=f" -U "
            f"{self.pri_userdb_pub.db_user} -W "
            f"{self.pri_userdb_pub.ssh_password} --remote-proto=ssh "
            f"--remote-host={self.pri_userdb_pub.db_host} "
            f"--remote-port=22 --remote-user={self.pri_userdb_pub.ssh_user} "
            f"--remote-path={self.remote_path} "
            f"--remote-libpath={self.remote_libpath} --backup-pg-replslot")
        self.assertTrue(result, '执行失败' + text)

        text = "--step8:使用全量备份远程恢复发布端主节点 expect:成功--"
        self.log.info(text)
        restore_mod = ''
        pub_result = self.commsh_pub.stop_db_cluster()
        self.log.info(pub_result)
        self.assertTrue(pub_result, '执行失败' + text)
        pub_result = self.commsh_pub_sta.exec_probackup_show(self.backup_path,
                                                             'instance2')
        result_split = pub_result.splitlines()
        self.log.info(result_split)
        for i in result_split:
            if 'FULL' in str(i):
                restore_mod = str(str(i).split()[2]).strip()
                self.log.info(restore_mod)
        pub_result = self.commsh_pub_sta.exec_pro_backup_restore(
            self.backup_path, 'instance2', restore_mod,
            '--incremental-mode=checksum --remote-proto=ssh '
            f'--remote-host={self.pri_userdb_pub.db_host} '
            f'--remote-port=22 --remote-user={self.pri_userdb_pub.ssh_user} '
            f'--remote-path={self.remote_path} '
            f'--remote-libpath={self.remote_libpath}')
        self.log.info(pub_result)
        self.assertTrue(pub_result, '执行失败' + text)
        pub_result = self.commsh_pub.start_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in pub_result or 'Degrade' in \
              pub_result
        self.assertTrue(flg, '执行失败' + text)
        time.sleep(5)
        for i in range(2):
            result = self.comsh_sta_pub[i].build_standby("-b full -t 3600")
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, result, '执行失败' +
                          text)
        pub_status = self.commsh_pub.get_db_cluster_status(param='detail')
        self.log.info(pub_status)
        self.assertEqual(pub_status.count('Normal'), 4, '执行失败' + text)

        text = "--step9:发布端插入表数据 expect:成功--"
        self.log.info(text)
        pub_result = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name1} values(55,1,1);"
            f"insert into {self.tb_name2} values(55,'first','%一','');")
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         '执行失败' + text)

        text = "--step10:订阅端查询数据是否更新 expect:数据更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);" \
            f"select * from {self.tb_name1};select * from {self.tb_name2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('55 |   1 |   1', sub_result, '执行失败' + text)
        self.assertIn('55 | first        | %一', sub_result, '执行失败' + text)

        text = "--step11:使用增量备份恢复远程恢复发布端主节点 expect:成功--"
        self.log.info(text)
        restore_mod = ''
        pub_result = self.commsh_pub.stop_db_cluster()
        self.log.info(pub_result)
        self.assertTrue(pub_result, '执行失败' + text)
        pub_result = self.commsh_pub_sta.exec_probackup_show(self.backup_path,
                                                             'instance2')
        result_split = pub_result.splitlines()
        for i in result_split:
            if 'PTRACK' in str(i):
                restore_mod = str(str(i).split()[2]).strip()
        pub_result = self.commsh_pub_sta.exec_pro_backup_restore(
            self.backup_path, 'instance2', restore_mod,
            '--incremental-mode=checksum --remote-proto=ssh '
            f'--remote-host={self.pri_userdb_pub.db_host} '
            f'--remote-port=22 --remote-user={self.pri_userdb_pub.ssh_user} '
            f'--remote-path={self.remote_path} '
            f'--remote-libpath={self.remote_libpath}')
        self.log.info(pub_result)
        self.assertTrue(pub_result, '执行失败' + text)
        pub_result = self.commsh_pub.start_db_cluster(True)
        flg = self.constant.START_SUCCESS_MSG in pub_result or 'Degrade' in \
              pub_result
        self.assertTrue(flg, '执行失败' + text)
        time.sleep(5)
        for i in range(2):
            result = self.comsh_sta_pub[i].build_standby("-b full -t 3600")
            self.assertIn(self.constant.BUILD_SUCCESS_MSG, result, '执行失败' +
                          text)
        pub_status = self.commsh_pub.get_db_cluster_status(param='detail')
        self.log.info(pub_status)
        self.assertEqual(pub_status.count('Normal'), 4, '执行失败' + text)

        text = "--step12:发布端插入表数据 expect:成功--"
        self.log.info(text)
        pub_result = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name1} values(99,1,1);"
            f"insert into {self.tb_name2} values(99,'first','%一','');")
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         '执行失败' + text)

        text = "--step13:订阅端查询数据是否更新 expect:表1更新,表2不更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('99 |   1 |   1', sub_result, '执行失败' + text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertNotIn('99 | first        | %一', sub_result, '执行失败' +
                         text)

        text = "--step14:发布端删除表数据 expect:成功--"
        self.log.info(text)
        pub_result = self.commsh_pub.execut_db_sql(
            f"delete from {self.tb_name1};delete from {self.tb_name2};")
        self.log.info(pub_result)
        self.assertEqual(pub_result.count(self.constant.DELETE_SUCCESS_MSG), 2,
                         '执行失败' + text)

        text = "--step15:订阅端查询数据是否更新 expect:表1更新,表2不更新--"
        self.log.info(text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name1};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('(1 row)', str(sub_result.splitlines()[-1]).strip(),
                      '执行失败' + text)
        sql_select = f"select pg_sleep(5.5);select * from {self.tb_name2};"
        sub_result = self.commsh_sub.execut_db_sql(
            sql_select, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_result)
        self.assertIn('(2 rows)', str(sub_result.splitlines()[-1]).strip(),
                      '执行失败' + text)

    def tearDown(self):
        text = "--step16:清理环境 expect:成功--"
        self.log.info(text)
        sql_cmd = f"rm -rf {self.backup_path};ls -al {self.backup_path};"
        self.log.info(sql_cmd)
        pub_result1 = self.pri_userdb_pub_sta.sh(sql_cmd).result()
        self.log.info(pub_result1)
        sql_cmd = f"drop subscription if exists {self.subname};"
        drop_sub_result = self.commsh_sub.execut_db_sql(
            sql_cmd, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_result)
        sql_cmd = "select * from pg_replication_slots;"
        pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(pub_result)
        sql_cmd = f"drop publication if exists {self.pubname};"
        drop_pub_result = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(drop_pub_result)
        sql_cmd = f"drop table if exists {self.tb_name1};" \
            f"drop table if exists {self.tb_name2};"
        result_sub = \
            self.commsh_sub.execut_db_sql(sql_cmd, self.user_param_sub, None,
                                          macro.DB_ENV_PATH_REMOTE1)
        self.log.info(result_sub)
        result_pub = self.commsh_pub.execut_db_sql(
            sql_cmd, sql_type=self.user_param_pub)
        self.log.info(result_pub)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
        self.log.info(cmd)
        sh_result = self.pri_userdb_pub.sh(cmd).result()
        self.log.info(sh_result)
        cmd = f"mv " \
            f"{os.path.join(self.parent_path_sub, 'pg_hba.conf')} "\
            f"{os.path.join(macro.DB_INSTANCE_PATH_REMOTE1, 'pg_hba.conf')} "
        self.log.info(cmd)
        sh_result1 = self.pri_userdb_sub.sh(cmd).result()
        self.log.info(sh_result1)
        result1 = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(result1)
        result2 = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'enable_cbm_tracking={self.enable_cbm_tracking}')
        self.log.info(result2)
        pub_result = self.commsh_pub.restart_db_cluster(True)
        sub_result = self.commsh_sub.restart_db_cluster(
            True, macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(f"No such file or directory", pub_result1,
                      '执行失败' + text)
        self.assertTrue(pub_result, '执行失败' + text)
        self.assertTrue(sub_result, '执行失败' + text)
        self.assertTrue(result1, '执行失败' + text)
        self.assertTrue(result2, '执行失败' + text)
        self.assertEqual("", sh_result, '执行失败' + text)
        self.assertEqual("", sh_result1, '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_result,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_result,
                      '执行失败' + text)
        self.assertEqual(result_sub.count(self.constant.DROP_TABLE_SUCCESS), 2,
                         '执行失败' + text)
        self.assertEqual(result_pub.count(self.constant.DROP_TABLE_SUCCESS), 2,
                         '执行失败' + text)
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
